package project;

/**
 * @author 鲁新茹
 * @date 2023/6/19
 */
/**
 * @author 张硕
 * @date 2023/6/18 16:24
 */
import org.apache.commons.lang.RandomStringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.text.DecimalFormat;
import java.util.Properties;
import java.util.Random;

import static java.lang.Double.parseDouble;

public class KafkaProducer1 {
    public static void main(String[] args) {
        // 1. 创建 kafka 生产者的配置对象
        Properties properties = new Properties();
        // 2. 给 kafka 配置对象添加配置信息：bootstrap.servers
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "niit01:9092");
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 每隔5秒推送 5条数据
        while (true) {
            // 3. 创建 kafka 生产者对象
            KafkaProducer<String, String> kafkaProducer = new KafkaProducer<String, String>(properties);
            System.out.println("开始发送数据");
            try {
                for (int i = 0; i <5; i++) {
                    Random rd = new Random();  //创建实例
                    int classNo = rd.nextInt(5)+202001;
                    String str=RandomStringUtils.random(8,new char[]{'a','b','c','d','e','f','g',
                            'h','i','j','k','l','m','n','o','p','q','r','s','t','u','v','w','x','y','z'});
                    //学期 1-8
                    int semester = rd.nextInt(8)+1;
                    // 0 1性别
                    int gender = rd.nextInt(2);

                    String[] birth = new String[]{"20020906","20020924","20010820","20000321","20010908","20030219","20020506","20020729","20011125"};
                    int bir=rd.nextInt(8);
                    String Birth = birth[bir];
                    int StudId = rd.nextInt(99) +2020340701;
                    String[] Statue = new String[]{"L", "A"};
                    int Stu = rd.nextInt(2);
                    String S = Statue[Stu];
                    //绩点
                    DecimalFormat decimalFormat = new DecimalFormat("0.00");
                    double GPA = parseDouble(decimalFormat.format(rd.nextDouble()*3.0+2.0));  //生成[2.0-5.0]区间的小数 保留两位，强制类型转换
                    // 品类
                    Object[] obj = new Object[8];  //Object数组可包含不同类型内容
                    obj[0] = classNo;obj[1] = str;
                    obj[2] = gender;obj[3] = Birth;
                    obj[4] = StudId;
                    obj[5] = semester;obj[6] = GPA;
                    obj[7] = S;

                    System.out.println("发送数据为" + obj[0]+"\t"+obj[1]+"\t"+obj[2]+"\t"+obj[3]+"\t"+obj[4]+"\t"+obj[5]+"\t"+obj[6]+"\t"+obj[7]);

                    try {

                        kafkaProducer.send(new ProducerRecord<String, String>("stuInfo3", obj[0]+"\t"+obj[1]+"\t"+obj[2]+"\t"+obj[3]+"\t"+obj[4]+"\t"+obj[5]+"\t"+obj[6]+"\t"+obj[7]));
//                        kafkaProducer.send(new ProducerRecord<String, String>("test", category));

                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
                Thread.sleep(5000);
            }catch(InterruptedException e){
                e.printStackTrace();
            }
//             5. 关闭资源
            kafkaProducer.close();

        }

    }
}